﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using WJP.Logging;
using WJP.Queue.Protocols.Brokers;
using WJP.Queue.Scheduling;
using WJP.Queue.Serializing;
using WJP.Queue.Socketing;

namespace WJP.Queue.NameServer
{
    /// <summary>
    /// 集群管理
    /// </summary>
    public class ClusterManager
    {
        private readonly ConcurrentDictionary<string, Cluster> _clusterDict;
        private readonly NameServerController _nameServerController;
        private ILogger _logger;

        private IScheduleService ScheduleService { get; set; }

        private IJsonSerializer JsonSerializer { get; set; }

        private IBinarySerializer BinarySerializer { get; set; }

        public ClusterManager(NameServerController nameServerController)
        {
            _clusterDict = new ConcurrentDictionary<string, Cluster>();
            _nameServerController = nameServerController;
            _logger = LogManager.Logger;
        }

        /// <summary>
        /// 启动服务
        /// </summary>
        public void Start()
        {
            _clusterDict.Clear();
            ScheduleService.StartTask("ScanNotActiveBroker", ScanNotActiveBroker, 1000, 1000);
        }

        /// <summary>
        /// 停止服务
        /// </summary>
        public void Shutdown()
        {
            _clusterDict.Clear();
            ScheduleService.StopTask("ScanNotActiveBroker");
        }

        /// <summary>
        /// 移除不再活动的Broker，每1秒扫描一次
        /// </summary>
        private void ScanNotActiveBroker()
        {
            foreach(var cluster in _clusterDict.Values)
            {
                foreach(var brokerGroup in cluster.BrokerGroups.Values)
                {
                    var notActiveBrokers = new List<Broker>();
                    foreach(var broker in brokerGroup.Brokers.Values)
                    {
                        if (broker.IsTimeout(_nameServerController.Setting.BrokerInactiveMaxMilliseconds))
                            notActiveBrokers.Add(broker);
                    }
                    if(notActiveBrokers.Count > 0)
                    {
                        foreach(var broker in notActiveBrokers)
                        {
                            if (brokerGroup.Brokers.TryRemove(broker.BrokerInfo.BrokerName, out Broker removed))
                                LogManager.Logger.InfoFormat("Removed timeout broker, brokerInfo: {0}, lastActiveTime: {1}", JsonSerializer.Serialize(removed.BrokerInfo), removed.LastActiveTime);
                        }
                    }
                }
            }
        }
    }

    /// <summary>
    /// 服务信息
    /// </summary>
    class Broker
    {
        public BrokerInfo BrokerInfo { get; set; }
        public long TotalSendThroughput { get; set; }
        public long TotalConsumeThroughput { get; set; }
        public long TotalUnConsumedMessageCount { get; set; }

        public IList<TopicQueueInfo> TopicQueueInfoList = new List<TopicQueueInfo>();
        public IList<TopicConsumeInfo> TopicConsumeInfoList = new List<TopicConsumeInfo>();
        public IList<string> ProducerList = new List<string>();
        public IList<ConsumerInfo> ConsumerList = new List<ConsumerInfo>();
        public string ConnectionId { get; set; }
        public ITcpConnection Connection { get; set; }
        public DateTime LastActiveTime { get; set; }
        public BrokerGroup Group { get; set; }
        public DateTime FirstRegisteredTime { get; set; }

        public bool IsTimeout(double timeoutMilliseconds)
        {
            return (DateTime.Now - LastActiveTime).TotalMilliseconds >= timeoutMilliseconds;
        }
    }

    /// <summary>
    /// 服务组信息
    /// </summary>
    class BrokerGroup
    {
        public string GroupName { get; set; }
        public ConcurrentDictionary<string, Broker> Brokers = new ConcurrentDictionary<string, Broker>();
    }

    /// <summary>
    /// 集群信息
    /// </summary>
    class Cluster
    {
        public string ClusterName { get; set; }
        public ConcurrentDictionary<string, BrokerGroup> BrokerGroups = new ConcurrentDictionary<string, BrokerGroup>();
    }
}
